package com.town.garden.config;

import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONArray;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.town.garden.constant.GlobalConstant;
import com.town.garden.entiy.Result;
import com.town.garden.entiy.request.FaceAddRequest;
import com.town.garden.propertites.NvrProperties;
import com.town.garden.propertites.PulsarProperties;
import com.town.garden.service.NvrFaceService;
import com.town.garden.utils.PulsarInitUtil;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.*;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.http.HttpStatus;

import java.util.concurrent.TimeUnit;

/**
 * @Description
 * @E-mail cwm3412638@163.com
 * @Author cwm
 * @Date 2022/5/17
 */
@Configuration
@EnableConfigurationProperties(value = PulsarProperties.class)
@Slf4j
@AllArgsConstructor
public class PulsarConfig {

    private final NvrFaceService nvrFaceService;

    /**
     * 创建pulsar客户端对象
     *
     * @param properties
     * @return
     * @throws PulsarClientException
     */
    @Bean
    public PulsarClient pulsarClient(PulsarProperties properties, NvrProperties nvrProperties) throws PulsarClientException {
        PulsarClient pulsarClient = PulsarClient.builder()
                .serviceUrl(properties.getServiceUrl())
                //如果业务流量较大，可以调高ioThreads个数
                .ioThreads(properties.getIoThreads())
                .listenerThreads(properties.getListenerThreads())
                .memoryLimit(properties.getMemoryLimit(), SizeUnit.MEGA_BYTES)
                .operationTimeout(properties.getOperationTimeout(), TimeUnit.SECONDS)
                .connectionTimeout(properties.getConnectionTimeout(), TimeUnit.SECONDS)
                .build();
        //只会初始化一个订阅
        initConsumer(pulsarClient, properties, nvrProperties);
        return pulsarClient;
    }

    /**
     * 初始化工具类
     *
     * @param pulsarClient
     * @param properties
     * @return
     */
    @Bean
    public PulsarInitUtil pulsarInitUtil(PulsarClient pulsarClient, PulsarProperties properties) {
        return new PulsarInitUtil(pulsarClient, properties.getServiceUrl());
    }

    /**
     * 初始化订阅主题
     *
     * @param pulsarClient
     * @param properties
     * @param nvrProperties
     */
    private void initConsumer(PulsarClient pulsarClient, PulsarProperties properties, NvrProperties nvrProperties) throws PulsarClientException {
        log.info("nvr subscription pulsar start topic is :{}", nvrProperties.getMac() + properties.getTopics());
        Consumer<String> consumers = pulsarClient.newConsumer(Schema.STRING)
                .topic(nvrProperties.getMac() + properties.getTopics())
                .subscriptionName(IdUtil.fastSimpleUUID())
                .receiverQueueSize(1)
                .ackTimeout(1, TimeUnit.SECONDS)
                .subscriptionType(SubscriptionType.Shared)
                .messageListener((MessageListener<String>) (consumer, msg) -> {
                    try {
                        log.error("nvr revice app message is :{}", msg.getValue());
                        FaceAddRequest faceAddRequest = JSONUtil.toBean(msg.getValue(), FaceAddRequest.class);
                        if (StrUtil.isNotEmpty(faceAddRequest.getData().getFaceInfo().get(0).getMd5())) {
                            //撤销布控
                            Result result = nvrFaceService.faceRemove(faceAddRequest);
                            sendWeb(result, pulsarClient, properties, "撤销布控", nvrProperties.getMac() + GlobalConstant.RECEIVED_DOWN_NVR_DEL);
                        } else {
                            //下发nvr
                            Result result = nvrFaceService.addFace(faceAddRequest);
                            sendWeb(result, pulsarClient, properties, "下发", nvrProperties.getMac() + GlobalConstant.RECEIVED_DOWN_NVR);
                        }
                    } finally {
                        try {
                            consumer.acknowledge(msg);
                        } catch (PulsarClientException e) {
                            e.printStackTrace();
                        }
                    }
                }).subscribe();
    }

    /**
     * 回馈处理结果到前段
     *
     * @param result
     * @param pulsarClient
     * @param properties
     */
    public void sendWeb(Result result, PulsarClient pulsarClient, PulsarProperties properties, String msg, String topic) {
        JSONObject jsonObject = JSONUtil.parseObj(result.getData());
        JSONObject jsonObject1 = JSONUtil.parseObj(jsonObject.get("data"));
        JSONArray code = JSONUtil.parseArray(jsonObject1.get("Result"));
        JSONArray md5 = JSONUtil.parseArray(jsonObject1.get("MD5"));
        Integer anInt = code.getInt(0);
        if (HttpStatus.OK.value() == result.getCode()) {
            if (anInt == 0) {
                Result ok = Result.ok(msg + "成功", anInt + GlobalConstant.SPLIC_STR + md5.getStr(0));
                log.info("send face topic is :{}",topic);
                pulsarInitUtil(pulsarClient, properties).sendMsg(IdUtil.fastSimpleUUID(), JSONUtil.toJsonStr(ok), pulsarInitUtil(pulsarClient, properties).getProducer(topic));
            } else {
                Result fail = Result.fail(msg + "失败", anInt);
                pulsarInitUtil(pulsarClient, properties).sendMsg(IdUtil.fastSimpleUUID(), JSONUtil.toJsonStr(fail), pulsarInitUtil(pulsarClient, properties).getProducer(topic));
                log.error("send face is fail,fail reason :{}", code.getInt(0));
            }
        } else {
            Result fail = Result.fail(msg + "失败", anInt);
            pulsarInitUtil(pulsarClient, properties).sendMsg(IdUtil.fastSimpleUUID(), JSONUtil.toJsonStr(fail), pulsarInitUtil(pulsarClient, properties).getProducer(topic));
            log.error("send face is fail ,fail reason :{}", result.getMessage());
        }
    }
}
